package cn.edu.flink.tutorial.broadcast;

import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.accumulators.ListAccumulator;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.List;
import java.util.Map;

public class 累加器 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Long> streamSource = env.addSource(new MySourceFunction());

        streamSource.map(new RichMapFunction<Long, Object>() {
            // 创建累加器
            final ListAccumulator<Long> listAccumulator = new ListAccumulator<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                // 注册累加器
                getRuntimeContext().addAccumulator("listAccumulator", listAccumulator);
            }

            @Override
            public Object map(Long value) throws Exception {
                listAccumulator.add(value);
                return null;
            }
        });

        List listAccumulator = env.execute().getAccumulatorResult("listAccumulator");

        listAccumulator.forEach(System.out::println);
    }

    static class MySourceFunction implements SourceFunction<Long> {
        public static MySourceFunction getInstance() {
            return new MySourceFunction();
        }

        @Override
        public void run(SourceContext<Long> ctx) throws Exception {
            for (int i = 0; i < 10; i++) {
                ctx.collect(System.currentTimeMillis());
                Thread.sleep(500L);
            }
        }

        @Override
        public void cancel() {
            System.out.println("source cancel");
        }
    }
}

